package cn.tedu.rabbitmq.test;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

public class Producer {
    private Channel channel;
    @Before
    public void initChannel() throws IOException, TimeoutException {
        // 通过java客户端建立和rabbitmq的连接
        ConnectionFactory factory = new ConnectionFactory();
        // 给工厂提供连接属性 ip:port
        factory.setHost("192.168.8.100");
        factory.setPort(5672);
        // 权限账号 admin/123456
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 虚拟主机：每个连接工厂由一个程序管理，程序所属的环境 连接的虚拟主机virtual_host不同
        // 实现不同环境的数据隔离 默认就是/
        factory.setVirtualHost("/");
        Connection connection = factory.newConnection();
        // 可以从长连接获取短链接
        channel = connection.createChannel();
    }
    @Test
    public void sendMessage() throws IOException {
        for (int i = 0; i < 100; i++) {
            // 测试消息内容 可以是任何类型的对象数据,需要经过序列化编程byte数组
            String message="hello world" + i;
            /*
            * String exchange: 交换机名称,空字符串表示默认交换机
            * String routingKey: 路由键,交换机可以根据路由键值,计算对应的队列 转发消息
            * BasicProperties props: 消息属性. 比如消息优先级priority appId userId 关联
            * byte[] body: 消息体 消息内容
            * */
            channel.basicPublish("","queue1",null,message.getBytes());
            System.out.println(i + "消息发送成功");
        }
    }
    @Test
    public void sendMessage01() {
        String message="hello world01~";
        try{
            // 开启事务
            channel.txSelect();
            channel.basicPublish("", "queue1", null, (message).getBytes());
            // 开启事务之后,线程阻塞等待
            // 询问rabbtimq是否消息达到 如果到达 提交结束 没有到达异常反馈
            AMQP.Tx.CommitOk commitOk = channel.txCommit();
            System.out.println("消息发送成功:"+message);
        }catch (Exception e){
            // 记录日志 调用 告警接口
        }
    }
    @Test
    public void sendMessage02() throws IOException, InterruptedException {
        String message="hello world02~";
        //确认模式 异步回调通知结果
        //开启确认模式
        channel.confirmSelect();
        channel.basicPublish("", "queue1", null, (message).getBytes());
        boolean flag = channel.waitForConfirms();
        if(flag){
            System.out.println("消息发送成功:"+message);
        }
    }
    @Test
    public void sendMessage03() throws IOException {
        String message="hello world03~";
        channel.basicPublish("direct_01","南京",null, message.getBytes());
    }
    @Test
    public void sendMessage04() throws IOException {
        String message="hello world04~";
        channel.basicPublish("fanout_01","长三角",null, message.getBytes());
    }
    @Test
    public void sendMessage05() throws IOException {
        String message="hello world05~";
        channel.basicPublish(
                "topic_ex_02",
                "中国.南京.玄武区",
                null, message.getBytes());
    }
    @Test
    public void sendHeaderMessage() throws IOException {
        //在消息中携带头
        String message="消息";
        //携带消息的属性头
        Map<String, Object> headers = new HashMap<>();
        headers.put("user-id", "12345");
        headers.put("message-type", "order");
        headers.put("priority", 1);
        headers.put("timestamp", System.currentTimeMillis());

        // 5. 创建消息属性并设置头数据
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .headers(headers)
                .contentType("text/plain")
                .deliveryMode(2) // 持久化消息
                .build();
        channel.basicPublish(
                "",
                "queue1",properties,message.getBytes());
    }
}
